#include <UTILS/SBuffer.h>
#include <PECAgent/PEContext.h>
#include <DPS/PH/DpsFunctions.h>
#include <DPS/PH/DpsOPContext.h>
#include "UDOP_ReadCdump.h"
#include "iostream"
#include "fstream"
#include "utilities.h"
#include "rngs.h"

SPC_NAMESPACE_USE;
UTILS_NAMESPACE_USE;
using namespace DPS;
using namespace std;
using namespace estd;

/*
 * Function used to initialize the operator
 */
void UDOP_ReadCdump::initializeOperator() 
{
  // Create a user defined metric
  // opm.createMetric("intuples","Number of ingested tuples");
}

/*
 * Function used to finalize the operator
 */
void UDOP_ReadCdump::finalizeOperator() 
{
}

/*
 * Function used to process command line arguments
 */
void UDOP_ReadCdump::processCmdArgs(const string & /*args*/) 
{
  SPCDBG(L_INFO, "Process command line arguments...", "dpsop");
  // ArgContainer arg(args); // include "ArgContainer.h"
  // arg.argc and arg.argv are now available for use
  // do other initialization here
}


/*
 * Process function for this source operator
 */
void UDOP_ReadCdump::processInput0(const IPort0_t & tuple)
{
  int k = 0;
  int info = 0;
  int streamid = 0;
  int yr = 0; int cur_yr = 0;
  int mo = 0; int cur_mo = 0;
  int dy = 0; int cur_dy = 0;
  int hr = 0; int cur_hr = 0;
  float slat = 0.0;
  float slon = 0.0;
  double scon = 0.0;
//  double drand1 = 0.0;
//  double drand2 = 0.0;
  char cdumpfile[180];
  char str1[100];
  int ssid;
  bool flag = true;
  SPCDBG(L_DEBUG, "Process...", "dpsop");
  const DpsOPContext &context = this->getContext();
  const std::string &args = context.getCmdArguments();
  //SPCDBG(L_INFO, "args is " << args, "dpsop");
  strcpy(str1, args.c_str());
  sscanf(str1,"%*s %*c%d%*c %*s %*c%d%*c",&streamid,&astep);
  ssid = streamid - 1;
  info = std::sprintf(cdumpfile, "/home/shiming1/InfoSphereStreams/letkf-spade/hysplit/data/test%d", ssid);
  OPort0_t otuple;
  ifstream fin;
  fin.open(cdumpfile, ios::in);
  fin.getline(str1, 100);
  tstep++;
  // Get time ifo, and filter record basing on time
  get_time(&cur_yr, &cur_mo, &cur_dy, &cur_hr);
  
  dlist<float> lat = DPS::makeFloatList();
  dlist<float> lon = DPS::makeFloatList();
  dlist<double> con = DPS::makeDoubleList();

  while (!fin.eof())
	{
	fin>>yr;
	fin>>mo;
	fin>>dy;
	fin>>hr;
	fin>>slat;
	fin>>slon;
	fin>>scon;
	if ((yr == (cur_yr+1900)) && (mo == cur_mo) 
			&& (dy == cur_dy) && (hr == cur_hr))
		{
/*		if (tstep == 1)
		{//initial random guess
		drand1 = Random();
		drand2 = Random();
		slat = slat + (float)(0.005*(drand1 - drand2) );
		drand1 = Random();
		drand2 = Random();
		slon = slon + (float)(0.005*(drand1 - drand2) );
		drand1 = Random();
		drand2 = Random();
		scon = scon * (1.0 + 0.01*abs(drand1-drand2));
		}*/
			
		lat.push_back(slat);
		lon.push_back(slon);
		con.push_back(scon);
		k++;
		}
	}
	
	if ((k > 0) && (tstep%astep==0) )
	{
		flag = true; // There is observation
	}
	else
	{
		flag = false; // There is no observation
	}
	otuple.set_sid(ssid);
	otuple.set_lat(lat);
	otuple.set_lon(lon);
	otuple.set_con(con);
	otuple.set_flag(flag);
	
	fin.close();
	submit0(otuple);
}


/*
 * Function used to notify the operator about the readiness of all ports (this
 * is useful when submitting tuples from within a context other than the process
 * functions of the operator).
 */ 
/*
void UDOP_ReadCdump::allPortsReady() 
{
} 
*/

/*
 * Terminate function, called asynchronously
 */
/*
void UDOP_ReadCdump::prepareToTerminate() 
{
}
*/

/*
 * Function used to update performance counters
 */
/*
static void updatePerformanceCounters(DpsOPMetrics & opm) 
{ 
  static long long ntuples = 0;
  // This is not the ideal way of using performance counters
  // Ideally one should store the result from getMetricByName, 
  // in a class member variable. For instance:
  //   DpsMetric & m = opm.getMetricByName("intuples");
  // m could be made a member variable of UDOP_ReadCdump
  // Then, the following is a more efficient way to use the counter:
  //   m.setValue(ntuples++);
  opm.getMetricByName("intuples").setValue(ntuples++);
}
*/

/*
 * Checkpointing function for saving state, called asynchronously 
 */
/*
void UDOP_ReadCdump::getCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Getting operator state...", "dpsop");
}
*/

/*
 * Checkpointing function for restoring state, called asynchronously
 */
/*
void UDOP_ReadCdump::restoreCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Restoring operator state...", "dpsop");
}
*/


